package com.wzw.basics.concurrent.thread.future;

import com.wzw.basics.concurrent.thread.ThreadPool;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.*;
import java.util.stream.Collectors;

/**
 * @author Wangzhiwen
 */
public class CompletableFutureApi {

    public static final ThreadPoolExecutor FUTURE_POOL = ThreadPool.defaultParamsPoolExecutor("completableFuture");

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        demo();
    }

    public static void api() throws ExecutionException, InterruptedException {
        // 任务创建
        // 创建无返回值的异步任务
        CompletableFuture.runAsync(() -> System.out.println("CompletableFuture 执行无返回值的异步任务"));

        // 无返回值，可指定线程池（默认使用ForkJoinPool.commonPool）
        CompletableFuture.runAsync(() -> System.out.println("CompletableFuture 带线程池 执行无返回值的异步任务"), FUTURE_POOL);

        // 创建有返回值的异步任务
        CompletableFuture<String> supplyAsync1 = CompletableFuture.supplyAsync(() -> "completableFuture 执行有返回值的异步任务");
        if (supplyAsync1.isDone()) {
            System.out.println(supplyAsync1.get());
        }

        // 有返回值，可指定线程池
        CompletableFuture<String> supplyAsync2 = CompletableFuture.supplyAsync(() -> "CompletableFuture 带线程池 执行有返回值的异步任务", FUTURE_POOL);
        if (supplyAsync2.isDone()) {
            // 通过future.get()方法获取异步任务的结果，还是会阻塞主线程的等待任务完成
            System.out.println(supplyAsync2.get());
        }

        // 执行回调 不阻塞主线程，在异步任务完成后自动执行回调方法中的代码
        // 方法执行第二个任务时都是异步执行
        // thenRun(); 与第一个方法公用一个线程池
        // thenRunAsync(Runnable action); 使用默认的线程池
        // thenRunAsync(Runnable action,Executor executor); 使用传入的线程池

        // 回调执行，无参数无返回值
        supplyAsync2.thenRun(() -> System.out.println("CompletableFuture then run"));

        // 回调执行，接受参数无返回值
        supplyAsync2.thenAccept(s -> System.out.println(s + " then accept"));

        // 回调执行，接受参数带返回值
        CompletableFuture<String> future3 = supplyAsync2.thenApply(s -> s + " then apply");

        // 回调执行，如果由异常会传入
        supplyAsync2.whenComplete((s, throwable) -> System.out.println("CompletableFuture when complete throwable"));

        // 组合异步任务
        // thenCompose() 顺序执行：第一个任务执行完后在执行里面任务
        // thenCombine() 并行执行：第一个任务与第二个任务并行执行
        supplyAsync2.thenCompose(s -> CompletableFuture.supplyAsync(() -> s + "顺序执行任务2"));

        CompletableFuture<String> task3 = CompletableFuture.supplyAsync(() -> "并行执行任务3");
        // 参数一：任务 参数二：任务一与任务二返回的值
        CompletableFuture<String> combine = supplyAsync2.thenCombine(task3, (s, s2) -> s + " " + s2);
        String combineResult = combine.get();
        System.out.println(combineResult);

        // 多任务组合
        // 全部完成时执行 无返回值
        CompletableFuture.allOf(create("task--1"), create("task--2")).join();
        // 任意一个完成时执行
        Object anyResult = CompletableFuture.anyOf(create("task--3"), create("task--4")).get();

        // 异常处理
        CompletableFuture<String> task5 = create("task--5");
        String task5Result = task5.exceptionally(throwable -> {
            // 处理异常
            throwable.printStackTrace();
            // 异常时需要返回的值
            return "default";
        }).get();

        // 后置处理
        CompletableFuture<String> task6 = create("task--6");
        // 传入线程返回的结果、出现的异常、处理后需要返回的结果类型
        Long handleResult = task6.handle(new BiFunction<String, Throwable, Long>() {
            @Override
            public Long apply(String s, Throwable throwable) {
                return 9999999L;
            }
        }).get();

        // 分片执行 借助stream实现，然后通过CompletableFuture实现并行执行，最后做数据聚合.
        List<CompletableFuture<String>> fragmentation = new ArrayList<>();
        for (int i = 0; i <= 10; i++) {
            fragmentation.add(create("task--" + i));
        }
        List<String> aggregation = fragmentation.stream().map(CompletableFuture::join).collect(Collectors.toList());
    }

    public static CompletableFuture<String> create(String result) {
        return CompletableFuture.supplyAsync(() -> result);
    }

    public static void demo() {
        // 任务1
        CompletableFuture<String> task1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("执行任务1");
            sleep(2);
            return "任务1";
        });
        // 任务2
        CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("执行任务2");
            sleep(4);
            return "任务2";
        });
        // 任务3 任务1与任务2并行执行 任务3等待
        CompletableFuture<String> task3 = task1.thenCombine(task2, (s, s2) -> {
            System.out.println("执行任务3");
            return s + " " + s2;
        });
        String join = task3.join();
        System.out.println(join);
    }

    private static void sleep(int i) {
        try {
            TimeUnit.SECONDS.sleep(i);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}
